package com.code.soulnetty.modbus.client.channel.impl;

import com.code.soulnetty.modbus.client.message.ProtocolRequest;
import com.code.soulnetty.modbus.client.message.ProtocolRequestWrap;
import com.code.soulnetty.modbus.client.message.ProtocolResponse;
import com.code.soulnetty.modbus.client.service.ProtocolResponseService;
import com.google.common.collect.Maps;
import com.sun.istack.internal.NotNull;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static com.code.soulnetty.modbus.biz.ThreadPoolUtils.bus;

/**
 * @author lyf
 * @date 2022-12-17 11:59
 */
@Slf4j
public class NettyAsyncChannel implements ProtocolChannel {

    private static final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(8, new ThreadFactory() {
        AtomicInteger ind = new AtomicInteger();

        @Override
        public Thread newThread(@NotNull Runnable r) {
            return new Thread(r, "modbus." + ind.incrementAndGet());
        }
    });

    private SocketAddress remoteSocketAddress;
    // 抽象的通道接口
    private final Channel channel;
    // 请求映射数据
    private final Map<String, ProtocolRequest> requestMap = Maps.newConcurrentMap();
    // 响应映射数据
    private final Map<String, CompletableFuture<ProtocolResponse>> futureMap = Maps.newConcurrentMap();
    // 响应数据的 业务处理类
    private ProtocolResponseService responseService;
    // 重发任务容器
    private final Map<String, ScheduledFuture<?>> schedules = new ConcurrentHashMap<>();

    // 上个数据周期
    private volatile int lastPeriod;
    // 重试次数
    private int retryCount;
    // 超时时间
    private long timeout;
    // 轮训间隔时间
    private long reqIntl;

    public NettyAsyncChannel(SocketAddress remoteSocketAddress, Channel channel, ProtocolResponseService responseService,
                             int retryCount, long timeout, long reqIntl) {
        this.remoteSocketAddress = remoteSocketAddress;
        this.channel = channel;
        this.responseService = responseService;
        this.retryCount = retryCount;
        this.timeout = timeout;
        this.reqIntl = reqIntl;
    }

    @Override
    public SocketAddress getRemoteAddress() {
        return remoteSocketAddress;
    }

    @Override
    public SocketAddress getLocalAddress() {
        return channel.localAddress();
    }

    @Override
    public boolean isConnected() {
        return channel.isActive();
    }

    protected Map<String, ProtocolRequest> getRequestMap() {
        return this.requestMap;
    }

    @Override
    public CompletableFuture<ProtocolResponse> send(ProtocolRequest request) throws Exception {
        // 策略
        Pair<Boolean, String> msgIdInfoPair = request.msgIdInfo();
        CompletableFuture<ProtocolResponse> future = new CompletableFuture<>();
        this.futureMap.put(msgIdInfoPair.getRight(), future);
        send((ProtocolRequestWrap) request, msgIdInfoPair.getRight());
        return future;
    }

    private void send(ProtocolRequestWrap request, String msgId) throws Exception {
        if (request.getPeriod() > lastPeriod) {
            lastPeriod = request.getPeriod();
        }
        final byte[] data = request.requestByteData();
        requestMap.put(msgId, request);
        try {
            final ChannelFuture channelFuture = channel.writeAndFlush(data).sync().addListener(future -> {
                if (this.retryCount > 1 && isConnected()) {
                    // 循环递归调用，来实现发送重试
                    ScheduledFuture<?> schedule = executor.schedule(() -> retrySend(request, msgId), this.timeout, TimeUnit.MILLISECONDS);
                    schedules.put(msgId, schedule);
                }
            });
            channelFuture.await(100, TimeUnit.MILLISECONDS);
            final Throwable cause = channelFuture.cause();
            if (cause != null) {
                throw new RuntimeException("modbus", cause);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } catch (Exception e) {
            throw new Exception(e);
        }
    }

    private void retrySend(ProtocolRequestWrap request, String msgId) {
        //已经接收到响应消息
        if (!this.requestMap.containsKey(msgId)) {
            return;
        }
        if (request.getRetryCount() > this.retryCount) {
            // 删除
            clear(msgId);
            return;
        }
        // 重试次数累加
        request.setRetryCount(request.getRetryCount() + 1);
        try {
            send(request, msgId);
        } catch (Exception e) {
            log.error("retry send data error {},{}", getRemoteAddress(), e.getMessage());
        }
    }

    @Override
    public void received(ProtocolResponse response) {
        ProtocolRequestWrap request = (ProtocolRequestWrap) requestMap.get(response.getMsgId());
        CompletableFuture<ProtocolResponse> future = futureMap.get(response.getMsgId());
        if (future != null) {
            future.complete(response);
        }
        clear(response.getMsgId());

        if (request == null) {
            log.info("[CHANNEL] [OUT_RECEIVED] [{}][{}]", getRemoteAddress(), response.getMsgId());
            return;
        }

        // 非周期性请求, 不对结果数据进行上传处理
        if (request.getPeriod() == ProtocolRequestWrap.NON_PERIOD) {
            return;
        }

        // 判断当前请求是否全部返回结果
        if (requestMap.isEmpty()) {
            log.debug("[RECEIVED] [DONE] {} received done at {} and period is {}", channel.remoteAddress(),
                    System.currentTimeMillis(), request.getPeriod());
        } else if (System.currentTimeMillis() / 1000 - request.getPeriod() > 10) {
            log.warn("[RECEIVED] [NO] {} not done at {} and period is {}", channel.remoteAddress(),
                    System.currentTimeMillis(), request.getPeriod());
        }

        bus().submit(() -> responseService.handler(request, response));
    }

    private void clear(String msgId) {
        try {
            requestMap.remove(msgId);
            futureMap.remove(msgId);
            ScheduledFuture<?> remove = schedules.remove(msgId);
            if (remove != null) {
                remove.cancel(true);
            }
        } catch (Exception e) {
            log.error("retry send data clear error", e);
        }
    }

    @Override
    public void close() {
        try {
            // future.await() 阻塞到真正close掉，线程才会继续往下走
            channel.close().addListener(future -> {
                log.info("channel close {}", future.isSuccess());
                requestMap.clear();
                futureMap.clear();
            }).await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("channel close error.", e);
        }
    }

    @Override
    public void close(int timeout) {
        try {
            // future.await(timeout) 阻塞线程，若在timeout期间close完成则返回；否则等待timeout时间后，发现还没close，则立即返回往下执行
            channel.close().addListener(future -> {
                log.info("channel close {}, timeout: {}", future.isSuccess(), timeout);
                requestMap.clear();
                futureMap.clear();
            }).await(timeout);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("channel close error. timeout: {}", timeout, e);
        }
    }

    @Override
    public ProtocolRequest request(String msgId) {
        return requestMap.get(msgId);
    }

    @Override
    public int getRetryCount() {
        return this.retryCount;
    }

    @Override
    public long getTimeout() {
        return this.timeout;
    }

    @Override
    public long getReqIntl() {
        return this.reqIntl;
    }
}
